// Contain
package Pigeon

import (
	"encoding/json"
	// "flag"
	"fmt"
	"github.com/astaxie/beego/httplib"
	"github.com/golang/glog"
	"io"
	"io/ioutil"
	"time"
)

var (
	mqList map[string]*MqConfig
)

type MqConfig struct {
	url       string
	lastTime  int //消息存活时间
	delayTime int //延迟时间
	retryTime int //最多尝试时间
}

type MqContent struct {
	Qname        string //队列名称
	Content      string //内容
	AddTime      int64  //添加时间
	LastTime     int64  //上次取出时间
	HasRetryTime int    //已经尝试次数
}

type Result struct {
	Code   int         `json:"code"`
	Msg    string      `json:"msg"`
	Result interface{} `json:"result"`
}

func init() {
	mqList = make(map[string]*MqConfig)
}

func Run() {
	glog.Infoln("mqs starting")
	go func() {
		for {
			ConsumeMessage()
		}
	}()
	Server()
}

func ConsumeMessage() {

	contentByteArr, err := Get()
	if err == nil {
		if len(contentByteArr) > 1 {
			var myContent = new(MqContent)
			err := json.Unmarshal(contentByteArr, myContent)
			if err != nil {
				glog.Errorln("Unmarshal ", string(contentByteArr), " error", err.Error())
			} else {
				nowTimeStrampInt64 := time.Now().Unix()
				if mqConfPtr, ok := mqList[myContent.Qname]; ok {
					if int(nowTimeStrampInt64-myContent.LastTime) > mqConfPtr.delayTime {
						glog.Infoln("消费 message begin ", myContent.Qname, myContent.Content)
						pipMessage(myContent)
					} else {
						pushMessage(myContent)
					}
				} else {
					glog.Errorln("qname 不存在", myContent.Qname)
				}

			}
		}

	}
}

func AddMessage(qname, content string) error {
	nowTimeStrampInt64 := time.Now().Unix()
	myContent := new(MqContent)
	myContent.Content = content
	myContent.AddTime = nowTimeStrampInt64
	myContent.LastTime = nowTimeStrampInt64
	myContent.HasRetryTime = 0
	myContent.Qname = qname
	err := pushMessage(myContent)
	return err
}

func pushMessage(myContent *MqContent) error {
	contentJson, err := json.Marshal(myContent)
	if err != nil {
		glog.Errorln("json 消息错误")
		return nil
	} else {
		err := Set(contentJson)
		return err
	}
}

func pipMessage(content *MqContent) {
	nowTimeStrampInt64 := time.Now().Unix()

	thisMqConfig, ok := mqList[content.Qname]

	if ok == false {
		glog.Errorln("mq 不存在", content.Qname)
	} else {
		defer func() {
			if err := recover(); err != nil { //发生了错误push进去队列等待下次轮循
				if content.AddTime+int64(thisMqConfig.lastTime) > nowTimeStrampInt64 && content.HasRetryTime < thisMqConfig.retryTime {
					content.HasRetryTime++
					content.LastTime = nowTimeStrampInt64
					pushMessage(content)
				} else {
					glog.Errorln("消费失败", content)
				}
			}
		}()
		req := httplib.Post(thisMqConfig.url)
		message := content.Content
		var keyVal map[string]string
		err := json.Unmarshal([]byte(message), &keyVal)
		if err != nil {
			glog.Errorln("Unmarshal message err ", string(message))
		} else {
			for k, v := range keyVal {
				req.Param(k, v)
			}
			ret, err := req.Response()
			defer ret.Body.Close()
			if err != nil {
				glog.Errorln("Req error", err.Error())
				panic("Req error")
			} else {
				deaResult(ret.Body)
			}
		}
	}
}

func deaResult(readCloser io.ReadCloser) {
	myresult := new(Result)
	body, err := ioutil.ReadAll(readCloser)
	if err != nil {
		glog.Errorln(err.Error())
		panic("ReadAll error")
	}
	err = json.Unmarshal([]byte(body), myresult)
	if err != nil {
		glog.Errorln("Unmarshal json err", string(body))
		panic("消费失败")
	} else {
		if myresult.Code != 200 {
			err := fmt.Errorf("消费失败%v", myresult.Result)
			glog.Errorln(err.Error())
			panic("消费失败")
		} else {
			glog.Infoln("消费成功", string(body))
		}
	}
}

func AddMqList(mqName, mqUrl string, delayTime int, retryTime int, lastTime int) {
	myConfig := new(MqConfig)
	myConfig.url = mqUrl
	myConfig.delayTime = delayTime
	myConfig.retryTime = retryTime
	myConfig.lastTime = lastTime
	mqList[mqName] = myConfig
	glog.Infoln("添加Mq成功", mqName, mqList[mqName])
}

func RemoveMqList(mqName string) {
	delete(mqList, mqName)
}
